package com.gitee.ykr.lightweight.rpc.protocol.client;

import com.gitee.ykr.lightweight.rpc.common.container.ExcutorContainer;
import com.gitee.ykr.lightweight.rpc.common.entity.HeartBeat;
import com.gitee.ykr.lightweight.rpc.common.entity.NettyRpcRequest;
import com.gitee.ykr.lightweight.rpc.common.entity.NettyRpcResponse;
import com.gitee.ykr.lightweight.rpc.common.thread.ThreadPoolFactory;
import io.netty.channel.ChannelException;
import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.SimpleChannelInboundHandler;
import io.netty.handler.timeout.IdleStateEvent;
import io.netty.util.ReferenceCountUtil;

import java.net.SocketAddress;
import java.util.concurrent.Callable;

/**
 * @author YKR
 * @Describe
 * @date 2021/8/24
 */
public class NettyClientHandler extends SimpleChannelInboundHandler<NettyRpcResponse> {
    private ExcutorContainer excutorContainer;
    //是否开启心跳检测日志（先用sout代替），默认开启
    private boolean logHeartBeat = true;

    public NettyClientHandler(ExcutorContainer excutorContainer) {
        this.excutorContainer = excutorContainer;
    }


    /**
     * 消息监控，监听服务器传来的消息
     * @param ctx
     * @param response
     * @throws Exception
     */
    @Override
    protected void channelRead0(ChannelHandlerContext ctx, final NettyRpcResponse response) throws Exception {

        //通过线程池工厂创建客户端线程池
        ThreadPoolFactory.createClientPoolExecutor( response.getInterfaze() ).submit(new Callable<NettyRpcResponse>() {

            @Override
            public NettyRpcResponse call() throws Exception {


                try {
                    ResponseHandler responseHandler = excutorContainer.getResponseHandler();
                    responseHandler.handle( response );
                } catch (Exception e) {
                    System.out.println(String.format("Client handle failed", e));
                } finally {
                    ReferenceCountUtil.release( response );
                }
                return null;

            }
        } );

    }

    /**
     * 如果服务端没有心跳检测机制，如果异常断网（如拔网线）
     * @param context
     * @param cause
     * @throws Exception
     */
    @Override
    public void exceptionCaught(ChannelHandlerContext context, Throwable cause) throws Exception {

        if (cause instanceof ChannelException) {
            System.out.println(String.format("消息通道将被关闭，原因：{}", cause.getClass().getName()));
            context.close();
        }
        //TODO 发布异常处理事件
    }

    /**
     * 客户端监听写事件。也就是设置时间内没有与服务端交互则发送心跳信息给服务端
     * @param context
     * @param o
     * @throws Exception
     */
    @Override
    public void userEventTriggered(ChannelHandlerContext context, Object o) throws Exception {
        // 读写空闲时,发送心跳信息
        if (o instanceof IdleStateEvent) {
            /*IdleStateEvent event = (IdleStateEvent) e;
            IdleState state = event.state();
            if (state == IdleState.WRITER_IDLE) {*/
            NettyRpcRequest request = new NettyRpcRequest();
            request.setHeatBeat( true );
            request.setInterfaze( HeartBeat.class.getName() );
            request.setMethod( "beat" );
            request.setSyn( false );

            if (logHeartBeat) {
                System.out.println(String.format("发送心跳信息..."));
            }

            context.writeAndFlush( request );
        }

    }

    public SocketAddress getLocalAddress(ChannelHandlerContext context) {
        return context.channel().localAddress();
    }

    public SocketAddress getRemoteAddress(ChannelHandlerContext context) {
        return context.channel().remoteAddress();
    }
}

